package com.charles.bigdata;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.util.Utils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import com.charles.bigdata.common.spark.SparkJob;
import com.charles.bigdata.common.utils.SpringBootBeanUtils;
import com.charles.bigdata.config.SparkConfig;

/**
 * 程序启动入口
 * <p>
 * 原理是参数传入类路径通过反射获取类信息，并且使用到了springboot的implements CommandLineRunner让容器启动完成的时候执行。
 */
@SpringBootApplication
public class SparkApplication implements CommandLineRunner {

	@Autowired
	private SparkConfig sparkConfig;

	public static void main(String[] args) {
		SpringApplication.run(SparkApplication.class, args);
	}

	@Override
	public void run(String... args) throws Exception {
		//spark.executor.memory： executor memory是每个节点上占用的内存。每一个节点可使用内存
		//spark.executor.cores：顾名思义这个参数是用来指定executor的cpu内核个数，分配更多的内核意味着executor并发能力越强，能够同时执行更多的task
		//spark.cores.max：为一个application分配的最大cpu核心数，如果没有设置这个值默认为spark.deploy.defaultCores 估计就是这个节点的最大核心数

		// 初始化Spark环境
		SparkConf sparkConf = new SparkConf().setAppName(sparkConfig.getAppName()).setMaster(sparkConfig.getMaster())
				.set("spark.driver.host", sparkConfig.getSparkHome()).set("spark.executor.memory", "1G")
				.set("spark.cores.max", "6");
		// .set("spark.executor.cores", "6");

		// 解决cannot assign instance of java.lang.invoke.SerializedLambda问题
		// 将 jar上传到服务器
		//		sparkConf.setJars(new String[] { "d:\\spring-boot-spark-statistics-1.0.0-SNAPSHOT.jar",
		//				"d:\\spring-boot-spark-common-1.0.0-SNAPSHOT.jar" });

		System.setProperty("hadoop.home.dir", "D:\\lihui\\tools\\hadoop-common-2.7.4-bin-master");

		JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);

		// sparkJob的类路径
		String className = args[0];
		System.out.println("==================== sparkJob的类路径 : " + className);
		Class clazz = Utils.classForName(className);
		Object sparkJob = SpringBootBeanUtils.getBean(clazz);
		if (sparkJob instanceof SparkJob) {
			((SparkJob) sparkJob).execute(javaSparkContext);
		}
	}

}
